package com.wtwd.wtwddevice.mq;


import com.wtwd.device.mqtentity.MessageBase;
import com.wtwd.device.properties.MqProperties;
import com.wtwd.device.protocol.AppProtocol;
import com.wtwd.device.protocol.WTProtocol;
import com.wtwd.device.util.StringUtils;
import com.wtwd.wtwddevice.listener.ReceiveMsgServiceContext;
import com.wtwd.wtwddevice.listener.deviceServerMsg.handler.IMsgHandler;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.List;

/**
 * 设备服务器消息监听
 */
@Slf4j
@Component
public class DeviceServerMsgMonitor {

    @Resource
    private MqProperties mqProperties;

    @Resource
    private ReceiveMsgServiceContext receiveMsgServiceContext;

    @PostConstruct
    public void init() {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(mqProperties.getDeviceMsgConsumeGroup());
        try {
            consumer.setNamesrvAddr(mqProperties.getNamesrvAddr());
            consumer.subscribe(mqProperties.getDeviceMsgProduceTopic(), mqProperties.getDeviceMsgTag());
            consumer.setMaxReconsumeTimes(mqProperties.getConsumeRetryTimes());
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    for (MessageExt messageBaseExt : msgs) {
                        MessageBase messageBase = MessageBase.deserialize(messageBaseExt.getBody());
                        log.info("topic:{},tag:{}接收到设备服务器消息MessageBases {}: {}-{}", mqProperties.getDeviceMsgProduceTopic(), mqProperties.getDeviceMsgTag(), messageBaseExt.getReconsumeTimes(), messageBase.toString());
                        //处理消息
                        String cmdName = "";
                        if (messageBase.getType() == 0) {
                            WTProtocol wtProtocol = messageBase.getWtProtocol();
                            if (wtProtocol != null) {
                                cmdName = wtProtocol.getCmdName();
                            }
                        } else if (messageBase.getType() == 1) {
                            AppProtocol appProtocol = messageBase.getAppProtocol();
                            cmdName = appProtocol.getCmdName();
                        }
                        if (StringUtils.isNotEmpty(cmdName)) {
                            IMsgHandler msgHandler = null;
                            try {
                                msgHandler = receiveMsgServiceContext.getIReceiverHandler(cmdName);
                                msgHandler.handleMsg(messageBase);
                            } catch (Exception e) {
                                log.error("处理设备服务器消息MessageBases{}|异常: {}", messageBase.toString(), e.getMessage());
                            }
                        }
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            consumer.start();
            log.info("成功订阅 :{}", mqProperties.getDeviceMsgProduceTopic());
        } catch (MQClientException e) {
            e.printStackTrace();
        }

    }


}
